package com.xjscrm.console.config;

import cn.hutool.core.util.IdUtil;
import com.google.common.collect.ImmutableMap;
import com.xjscrm.console.chat.cache.CacheHolder;
import com.xjscrm.console.chat.mq.consumer.MsgSendConsumer;
import com.xjscrm.console.mapper.WwGroupMemberMapper;
import com.xjscrm.console.service.customer.chat.ChatTalkRecordService;
import com.xjscrm.console.service.customer.chat.ChatTalkSessionService;
import com.xjscrm.console.service.customer.chat.ChatTalkSessionStaffRelService;
import com.xjscrm.console.service.customer.wwCustomer.WwCustomerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.StandardEnvironment;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * @author mingyang.chen create in 2024/3/29 17:09
 */
@Configuration
@Slf4j
public class MqConfig implements EnvironmentAware, InitializingBean {

    public static String mqTag;
    @Autowired
    WwCustomerService wwCustomerService;
    @Autowired
    WwGroupMemberMapper wwGroupMemberMapper;
    @Autowired
    private RocketMQProperties rocketMQProperties;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private ChatTalkSessionService chatTalkSessionService;
    @Autowired
    private ChatTalkSessionStaffRelService chatTalkSessionStaffRelService;
    @Autowired
    private ChatTalkRecordService chatTalkRecordService;
    @Autowired
    private CacheHolder cacheHolder;
    @Autowired
    private Environment env;
//    @Autowired
//    private OfflineConsumer offlineConsumer;

    @Override
    public void setEnvironment(Environment environment) {
        if (!(environment instanceof StandardEnvironment)) {
            return;
        }
        String hostAddress;
        try {
            hostAddress = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
        String port = environment.getProperty("server.port");
        StandardEnvironment standardEnvironment = (StandardEnvironment) environment;
        standardEnvironment.getPropertySources().addLast(new MapPropertySource("custom", ImmutableMap.of("mqTag", hostAddress + ":" + port)));
        mqTag = environment.getProperty("mqTag");
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        String nameServer = rocketMQProperties.getNameServer();

        String group = "xj_group_" + IdUtil.fastSimpleUUID();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);

        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //
        consumer.registerMessageListener(new MsgSendConsumer(chatTalkSessionService, chatTalkRecordService, cacheHolder, rocketMQTemplate, wwCustomerService, wwGroupMemberMapper, chatTalkSessionStaffRelService));
//        consumer.subscribe("XJ_WW_DISPATCH_HANDLE_TOPIC", "*");
        consumer.subscribe("XJ_WW_DISPATCH_HANDLE_TOPIC", MessageSelector.byTag(mqTag));
        consumer.start();

        log.info("路由消息消费者启动成功. 标签: [{}], 分组: [{}]", mqTag, group);

//
//        String offlineGroup = "xj_staff_offline_" + IdUtil.fastSimpleUUID();
//        DefaultMQPushConsumer offlineConsumer = new DefaultMQPushConsumer(offlineGroup);
//
//        offlineConsumer.setNamesrvAddr(nameServer);
//        offlineConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        offlineConsumer.setMessageModel(MessageModel.BROADCASTING);
//        //
//        offlineConsumer.registerMessageListener(this.offlineConsumer);
//        offlineConsumer.subscribe("xj_staff_offline_topic", MessageSelector.byTag(mqTag));
//        offlineConsumer.start();
//
//        log.info("下线消息消费者启动成功. 标签: [{}], 分组: [{}]", mqTag, offlineGroup);
    }
}
